package event

import (
	"errors"
	"fmt"
	"sync"
	"time"
	"yunj/pkg/yunj/core/log"
)

// Listener 监听器
type Listener struct {
	Id      string               // 唯一标识（不能为空）
	Handler func(data any) error // 处理函数（不能为空）
	IsAsync bool                 // 是否异步
	Timeout time.Duration        // 超时时间
}

// 监听器列表
var listeners = map[string][]*Listener{}

// 保护listeners的读写锁
var mu sync.RWMutex

// 订阅事件
func Subscribe(eventName string, eventListeners ...*Listener) error {
	// 监听器校验
	for _, el := range eventListeners {
		if el.Id == "" {
			return fmt.Errorf("事件[%s]监听器Id不能为空", eventName)
		}
		if el.Handler == nil {
			return fmt.Errorf("事件[%s]监听器Handler不能为空", eventName)
		}
		for _, l := range listeners[eventName] {
			if l.Id == el.Id {
				return fmt.Errorf("事件[%s]监听器[%s]已存在", eventName, el.Id)
			}
		}
	}
	listeners[eventName] = append(listeners[eventName], eventListeners...)
	return nil
}

// 取消订阅
func Unsubscribe(eventName string, listenerId string) {
	mu.Lock()
	defer mu.Unlock()

	eventListeners, exists := listeners[eventName]
	if !exists {
		return
	}

	newEventListeners := make([]*Listener, 0, len(eventListeners)-1)
	for _, el := range eventListeners {
		if el.Id == listenerId {
			continue
		}
		newEventListeners = append(newEventListeners, el)
	}
	listeners[eventName] = newEventListeners
}

// 发布事件（增强类型校验、超时控制和错误处理）
// 返回同步处理的错误
func Publish(eventName string, data any) error {
	mu.RLock()
	eventListeners := make([]*Listener, 0) // 复制当前监听器列表，避免发布过程中被修改
	if ls, ok := listeners[eventName]; ok {
		eventListeners = append(eventListeners, ls...)
	}
	mu.RUnlock()
	if len(eventListeners) == 0 {
		return nil
	}

	var syncErrors []error
	// 处理所有监听器
	for _, el := range eventListeners {
		// 包装处理函数，支持超时控制
		handleFunc := func() error {
			if el.Timeout > 0 {
				// 超时控制：使用带超时的goroutine
				errChan := make(chan error, 1)
				go func() {
					errChan <- el.Handler(data)
				}()
				select {
				case err := <-errChan:
					return err
				case <-time.After(el.Timeout):
					return fmt.Errorf("处理超时")
				}
			}
			// 无超时：直接执行
			return el.Handler(data)
		}

		if el.IsAsync {
			// 异步执行
			go func(l Listener) {
				err := handleFunc()
				logRecord(eventName, &l, data, err)
			}(*el) // 传递当前listener副本，避免循环变量引用问题
		} else {
			// 同步执行
			err := handleFunc()
			logRecord(eventName, el, data, err)
			if err != nil {
				syncErrors = append(syncErrors, err)
			}
		}
	}

	// 3. 合并同步错误
	if len(syncErrors) > 0 {
		return errors.Join(syncErrors...)
	}
	return nil
}

// 日志记录
func logRecord(eventName string, listener *Listener, data any, err error) {
	msg := fmt.Sprintf("事件[%s]监听器[%s]处理", eventName, listener.Id)
	if err != nil {
		msg = fmt.Sprintf("%s失败！", msg)
		log.Error(msg, err, map[string]any{"data": data})
	} else {
		msg = fmt.Sprintf("%s成功！", msg)
		log.Info(msg, map[string]any{"data": data})
	}
}
